package com.rem.rabbitmq.boot;

import com.rabbitmq.client.Channel;
import com.rem.rabbitmq.boot.Adelay.DelayQueueConfig;
import com.rem.rabbitmq.boot.Adelay.QueueConfig;
import com.rem.rabbitmq.boot.Bconfirm.ConfirmConfig;
import com.rem.rabbitmq.boot.CbackupExchange.BackupExchangeConfig;
import com.rem.rabbitmq.boot.Dpriority.PriotiryConfig;
import com.rem.rabbitmq.boot.Elazy.LazyConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;

/**
 * 队列消费者
 *
 * @author Rem
 * @date 2021-12-30
 */

@Slf4j
@Component
public class QueueConsumer {


    /**
     * 监听死信队列里的消息
     *
     * @param message
     */
    @RabbitListener(queues = QueueConfig.QD_DEAD_QUEUE)
    public void receiveD(Message message) {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到死信队列信息{}", LocalDateTime.now(), msg);

    }

    /**
     * 监听使用插件的延迟队列消息
     *
     * @param message
     */
    @RabbitListener(queues = DelayQueueConfig.DELAY_QUEUE)
    public void receiveDelay(Message message) {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到延迟队列信息{}", LocalDateTime.now(), msg);

    }

    /**
     * 发布确认队列监听
     *
     * @param message
     */
    @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)
    public void receiveConfirem(Message message) {
        String msg = new String(message.getBody());
        log.info("发布确认消费者收到的消息是:{}", msg);
    }

    /**
     * 报警队列监听
     *
     * @param message
     */
    @RabbitListener(queues = BackupExchangeConfig.WARING_QUEUE)
    public void receiveBackupWaring(Message message) {
        String msg = new String(message.getBody());
        log.info("报警队列收到的消息是:{}", msg);
    }

    /**
     * 优先级队列监听
     *
     * @param message
     */
    @RabbitListener(queues = PriotiryConfig.PRIOTIRY_QUEUE)
    public void receivePriotiry(Channel channel, Message message) {
        String msg = new String(message.getBody());
        log.info("优先级队列收到的消息是:{}", msg);
    }

    /**
     * 惰性队列
     *
     * @param message
     */
    @RabbitListener(queues = LazyConfig.LAZY_QUEUE)
    public void receiveLazy(Message message) {
        String msg = new String(message.getBody());
        log.info("惰性队列 的消息是:{}", msg);
    }


    /**
     * channel.basicReject(long,boolean); 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列。
     * @param channel
     * @param message
     * @throws IOException
     */
    @RabbitListener(bindings = {@QueueBinding(value = @Queue(value = "direct.queue"), exchange = @Exchange(value = "amq.direct"), key = "hello")})
    public void getDirectMessage(Channel channel, Message message) throws IOException {
        try {
            // 模拟执行任务
            Thread.sleep(1000);
            // 模拟异常
            int a = 1 / 0;
            // 确认收到消息，false只确认当前consumer一个消息收到，true确认所有consumer获得的消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                System.out.println("消息已重复处理失败,拒绝再次接收");
                // 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                System.out.println("消息即将再次返回队列处理");
                // requeue为是否重新回到队列，true重新入队
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }

}
